import java.io.*;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.*;
import java.util.concurrent.*;

//控制报文发送队列管理Map
public class ControlConnectionModule {
    public Map<String, CtrlSendQueue> SendQueue_map;
    private static final ControlConnectionModule INSTANCE = new ControlConnectionModule();//类加载时创建对象

    private ControlConnectionModule() {
        System.out.println("ControlConnectionModule单例已创建");
        SendQueue_map = new ConcurrentHashMap<>();
    }

    static public ControlConnectionModule getInstance() {
        return INSTANCE;
        //先通过ControlConnectionModule获取单例对象INSTANCE
        //再通过INSTANCE.SendQueue_map访问SendQueue_map
    }
}

class ConnUtils {
    //获取3个可以"监听"的ServerSocket
    public static ServerSocket[] getAvailableServerPort() {
        ServerSocket[] ss_arr = new ServerSocket[3];
        try {
            ss_arr[0] = new ServerSocket(0);//ServerSocket(0)表示系统自动分配可用端口号
            ss_arr[1] = new ServerSocket(0);
            ss_arr[2] = new ServerSocket(0);
        } catch (IOException e) {
            e.printStackTrace();
        }
        /*int port = 8500;//从8500端口开始向后搜索可用端口

        ServerSocket[] ss_arr = new ServerSocket[3];
        int count = 0;
        while (count < 3) {
            try {
                ServerSocket socket = new ServerSocket(port);
                ss_arr[count] = socket;//若有可用端口,则创建ServerSocket
                count++;
            } catch (IOException e) {
                port++;
            }
        }*/
        return ss_arr;
    }
}

//这个队列的消费者是控制报文发送线程,生产者是其余模块
//这个队列需要改进一下,改成"take()会阻塞,put()不阻塞"
class CtrlSendQueue {
    private LinkedList<byte[]> send_queue;
    private int capacity = 10000;//容量上限(双向队列容量上限)

    CtrlSendQueue() {
        send_queue = new LinkedList<>();
    }

    CtrlSendQueue(int capacity) {
        this.capacity = capacity;
        send_queue = new LinkedList<>();
    }

    public byte[] take() {
        synchronized (send_queue) {
            while (send_queue.isEmpty()) {
                //System.out.println("CtrlSendQueue队列为空,消费者线程等待");
                try {
                    send_queue.wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            //从队列头部获取消息并返回
            byte[] chunk = send_queue.removeFirst();
            //System.out.println("已消费消息");
            send_queue.notifyAll();//唤醒"因为队列已经满"而被put()函数阻塞的线程
            return chunk;
        }
    }

    public void put(byte[] chunk) {
        synchronized (send_queue) {
            while (send_queue.size() == capacity) {
                System.out.println("CtrlSendQueue队列已满,生产者线程等待");
                try {
                    send_queue.wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            send_queue.addLast(chunk);
            //System.out.println("已生产消息 {}");
            send_queue.notifyAll();
        }
    }
}

class ControlConnectionThread implements Runnable {
    Socket ctrl_conn = null;
    Socket[] file_conn = new Socket[3];
    int APmark = -1;//0为主动方,1为被动方
    BufferedInputStream bis;
    BufferedOutputStream bos;

    Thread ListenThread;
    Thread SendThread;

    public ControlConnectionThread(Socket ctrl_conn_, int APmark_) {
        this.ctrl_conn = ctrl_conn_;
        this.APmark = APmark_;
    }

    @Override
    public void run() {
        try {
            bis = new BufferedInputStream(ctrl_conn.getInputStream());
            bos = new BufferedOutputStream(ctrl_conn.getOutputStream());

            //建立文件收发连接
            if (APmark == 0) {
                //"主动方"开始监听这3个端口,然后给"被动方"发送3个可用的TCP端口号
                ServerSocket[] ss_arr = ConnUtils.getAvailableServerPort();

                //2.开始监听自己发送的3个TCP端口(使用Future模式)
                Thread t0 = new Thread(() -> {
                    try {
                        file_conn[0] = ss_arr[0].accept();
                        ss_arr[0].close();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                });
                t0.start();
                Thread t1 = new Thread(() -> {
                    try {
                        file_conn[1] = ss_arr[1].accept();
                        ss_arr[1].close();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                });
                t1.start();
                Thread t2 = new Thread(() -> {
                    try {
                        file_conn[2] = ss_arr[2].accept();
                        ss_arr[2].close();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                });
                t2.start();
                //3.给"被动方"发送3个可用的TCP端口号
                bos.write(NetChunkUtil.intToByteArray(12));
                bos.write(NetChunkUtil.intToByteArray(ss_arr[0].getLocalPort()));
                bos.write(NetChunkUtil.intToByteArray(ss_arr[1].getLocalPort()));
                bos.write(NetChunkUtil.intToByteArray(ss_arr[2].getLocalPort()));

                bos.flush();

                t0.join();
                t1.join();
                t2.join();

                System.out.println("0号连接: " + file_conn[0].getInetAddress() + " " + file_conn[0].getPort());
                System.out.println("1号连接: " + file_conn[1].getInetAddress() + " " + file_conn[1].getPort());
                System.out.println("2号连接: " + file_conn[2].getInetAddress() + " " + file_conn[2].getPort());
            } else if (APmark == 1) {
                //接收3个TCP端口号
                byte[] pba = null;//port_byte_arr 接收到的端口号byte数组,共12个元素,相当于3个int
                int[] ports = new int[3];//接收到的3个端口号
                while (true) {
                    pba = NetChunkUtil.recive(bis);
                    if (pba != null) {
                        break;
                    }
                }
                ports[0] = NetChunkUtil.byteArrayToInt(new byte[]{pba[0], pba[1], pba[2], pba[3]});
                ports[1] = NetChunkUtil.byteArrayToInt(new byte[]{pba[4], pba[5], pba[6], pba[7]});
                ports[2] = NetChunkUtil.byteArrayToInt(new byte[]{pba[8], pba[9], pba[10], pba[11]});

                System.out.println("接收到端口" + ports[0] + " " + ports[1] + " " + ports[2]);

                file_conn[0] = new Socket(ctrl_conn.getInetAddress(), ports[0]);
                file_conn[1] = new Socket(ctrl_conn.getInetAddress(), ports[1]);
                file_conn[2] = new Socket(ctrl_conn.getInetAddress(), ports[2]);

                System.out.println("0号连接: " + file_conn[0].isConnected() + " " + file_conn[0].getInetAddress() + " " + file_conn[0].getPort());
                System.out.println("1号连接: " + file_conn[0].isConnected() + " " + file_conn[1].getInetAddress() + " " + file_conn[1].getPort());
                System.out.println("2号连接: " + file_conn[0].isConnected() + " " + file_conn[2].getInetAddress() + " " + file_conn[2].getPort());
            }

//====创建资源======================================================================

            //创建"控制报文发送队列"并把队列交给"控制报文发送队列管理Map"
            CtrlSendQueue send_queue = new CtrlSendQueue();
            ControlConnectionModule.getInstance().SendQueue_map.put(ctrl_conn.getInetAddress().toString(), send_queue);

            //创建"文件发送队列"并把队列交给"文件发送队列管理Map"
            BlockingQueue<byte[]> send_file_queue = new LinkedBlockingQueue<>();
            SendFileQueueMap.getMap().put(ctrl_conn.getInetAddress().toString(), send_file_queue);

            //为每个file_conn创建BufferedOutputStream
            //并把file_conn交给 "SendFileConnInfoMap管理所有"文件发送连接"的map" 进行管理
            BufferedOutputStream[] bos_arr = new BufferedOutputStream[3];
            bos_arr[0] = new BufferedOutputStream(file_conn[0].getOutputStream());
            bos_arr[1] = new BufferedOutputStream(file_conn[1].getOutputStream());
            bos_arr[2] = new BufferedOutputStream(file_conn[2].getOutputStream());
            SendFileConnInfoMap.SFInfo sfinfo = new SendFileConnInfoMap.SFInfo(file_conn, bos_arr);
            System.out.print("初始化线程创建的sfinfo: ");
            System.out.println(sfinfo);
            SendFileConnInfoMap.getMap().put(ctrl_conn.getInetAddress().toString(), sfinfo);

//====开启线程-以及开启线程之后才能做的初始化操作======================================================================

            //开启一个"控制报文监听线程"和一个"控制报文发送线程"
            ListenThread = new Thread(new ListenMessageThread(), "CTRL_LT_" + ctrl_conn.getInetAddress().toString());
            ListenThread.start();
            SendThread = new Thread(new SendMessageThread(send_queue), "CTRL_ST_" + ctrl_conn.getInetAddress().toString());
            SendThread.start();

            //此时 "发送队列"与"发送线程"均建立完毕


            //通知心跳管理模块接管此控制报文连接
            //通过给 心跳管理模块的 sur_map写入新值(共写入3个值,Addr(作为map的key),SendQueue)
            //连接建立完毕,如果是主动方,把心跳状态设置为HB_sur, 如果是被动方,把心跳状态设置为HB_wait

            //连接建立完毕 丢弃"主动方,被动方标识"APmark(即设置为-1)
            //APmark = -1;
            SurvivalInformation info;
            if (APmark == 0) {
                info = new SurvivalInformation(ctrl_conn.getInetAddress().toString(), HBstate.HB_sur, send_queue);
            } else {
                info = new SurvivalInformation(ctrl_conn.getInetAddress().toString(), HBstate.HB_wait, send_queue);
            }
            HeartBeatModule.sur_map.put(ctrl_conn.getInetAddress().toString(), info);

            //创建文件接收线程(3条)
            //文件接收是一直收的(反正没数据的时候会阻塞,不占cpu), 只要对方发,时时刻刻都能收, 但是发送是动态调用的
            new Thread(new ReciveFileThread(file_conn[0]), "FILE_LT_0_" + ctrl_conn.getInetAddress().toString()).start();
            new Thread(new ReciveFileThread(file_conn[1]), "FILE_LT_1_" + ctrl_conn.getInetAddress().toString()).start();
            new Thread(new ReciveFileThread(file_conn[2]), "FILE_LT_2_" + ctrl_conn.getInetAddress().toString()).start();


        } catch (IOException | InterruptedException e) {
            e.printStackTrace();
        }
    }

    //这是一个内部类
    class ListenMessageThread implements Runnable {
        //[心跳信息]队列(单例)
        //功能: 把[心跳报文] 拆包为 [心跳消息] 存入此队列,交给HeartBeatModule处理
        //关联: ListenMessageThread线程是这个队列的 "生产者者"
        ConcurrentLinkedQueue<HeartBeatMessage> HBqueue = HeartBeatQueue.getInstance().queue;
        BlockingQueue<FCL_S_rcvqueue_item> receiver_queue = FCL_S_ReceiverQueue.getQueue();
        @Override
        public void run() {
            byte[] ba = null;//byte_arr 接收到的端口号byte数组,共12个元素,相当于3个int
            while (true) {
                try {
                    Thread.sleep(1);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                ba = NetChunkUtil.recive(bis);
                if(ba==null) {//-->ba有可能为null!!这时候是否就说明"连接断开"了呢?因为连接还存在的时候,不会为null,而是会阻塞
                    System.out.println(Thread.currentThread().getName() + "控制报文接收线程停止");
                    break;
                }
                //把监听到的报文分别送给各个模块(当然现在暂时还没有别的模块)
                int MessageType = MessageEncapsulationUtils.JudgeMessageType(ba);
                switch (MessageType) {
                    case 0://收到心跳包
                        int HB_ser_num = MessageEncapsulationUtils.HeartBeatMessage_unpack(ba);
                        HBqueue.add(new HeartBeatMessage(ctrl_conn.getInetAddress().toString(), HB_ser_num));
                        break;
                    case 1://接收到 其他客户端发来的"自己的文件持有表"
                        receiver_queue.add(new FCL_S_rcvqueue_item(ba, ctrl_conn.getInetAddress().toString()));
                        break;
                }

                for (int i = 0; i < ba.length; i++) {
                    System.out.print(ba[i]);
                }
                System.out.println();
            }
        }
    }

    class SendMessageThread implements Runnable {
        SendMessageThread(CtrlSendQueue q) {
            send_queue = q;
        }

        CtrlSendQueue send_queue;

        @Override
        public void run() {
            try {
                while (true) {
                    bos.write(send_queue.take());

                    //bos.write(NetChunkUtil.intToByteArray(2));
                    //bos.write(98);
                    //bos.write(99);
                    bos.flush();//发一个,flush一次,否则会等到TCP缓存区满了再发
                    Thread.sleep(1);
                }
            } catch (IOException | InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}